﻿#include "pch.h"
#include "ThreadpoolWork.h"
#include "AsyncLibImpl.h"

namespace AsyncLib
{

CThreadpoolWork::CThreadpoolWork() : m_nIdleThreads(0)
{
	m_nMinThreads = std::thread::hardware_concurrency();
	m_nMaxThreads = m_nMinThreads * 2;
	m_vecThreads.reserve(m_nMaxThreads);
}

CThreadpoolWork::~CThreadpoolWork()
{
	assert(m_deqTasks.empty());
	assert(m_vecThreads.empty());
}

void CThreadpoolWork::Init()
{
}

void CThreadpoolWork::Stop()
{
	// Lock to prevent wait_for after notify_all
	std::unique_lock lock(m_mutex);
	m_condVar.notify_all();
}

void CThreadpoolWork::Exit()
{
	// Wait for all threads to exit
	std::unique_lock lock(m_mutex);
	auto vecThreads = std::move(m_vecThreads);
	lock.unlock();

	for (auto tid : vecThreads)
	{
		pthread_join(tid, nullptr);
	}

	// Destruct pending tasks sequentially
	while (m_deqTasks.size())
	{
		m_deqTasks.pop_front();
	}
}

void CThreadpoolWork::PostTask(std::function<void()>&& fnTask)
{
	std::unique_lock lock(m_mutex);
	m_deqTasks.emplace_back(std::move(fnTask));

	if (m_deqTasks.size() <= m_nIdleThreads)
	{
		m_condVar.notify_one();
	}
	else if (m_vecThreads.size() < m_nMinThreads || (m_vecThreads.size() < m_nMaxThreads && m_deqTasks.size() - m_nIdleThreads > m_vecThreads.size() / 2))
	{
		// Less than the number of core threads, or the number of tasks is more than half of the number of threads, start a new thread
		pthread_t tid = 0;
		int err = pthread_create(&tid, nullptr, [](void *pThis)
		{
			((CThreadpoolWork *)pThis)->ThreadProc();
			return (void *)0;
		}, this);

		assert(err == 0);

		if (err == 0)
		{
			m_vecThreads.emplace_back(tid);
			m_nIdleThreads++;
		}
	}
}

void CThreadpoolWork::ThreadProc()
{
	g_asyncLib.SetInThreadPool();
	std::unique_lock lock(m_mutex);
	m_nIdleThreads--;

	while (g_asyncLib.IsRunning())
	{
		if (m_deqTasks.size())
		{
			auto fnTask = std::move(m_deqTasks.front());
			m_deqTasks.pop_front();
			lock.unlock();

			fnTask();
			fnTask = nullptr;

			lock.lock();
		}
		else
		{
			m_nIdleThreads++;
			auto status = m_condVar.wait_for(lock, std::chrono::seconds(10));
			m_nIdleThreads--;

			// Exit when there are too many idle threads
			if (status == std::cv_status::timeout && m_nIdleThreads >= m_nMinThreads)
			{
				std::erase(m_vecThreads, pthread_self());
				break;
			}
		}
	}
}

} // namespace AsyncLib
